package cn.itcast.flink.window.time;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
/**
* 窗口统计案例演示：滚动时间窗口（Tumbling Time Window），实时交通卡口车流量统计
*/
public class StreamTumblingTimeWindow_1 {


public static void main(String[] args) throws Exception {
// 1. 执行环境-env：流计算执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1) ;
// 2. 数据源-source：Socket接收数据
DataStreamSource<String> inputStream = env.socketTextStream("node1.itcast.cn", 9999);
/*
数据源：信号灯编号和通过该信号灯的车的数量
a,3
a,2
a,7
d,9
b,6
a,5
b,3
e,7
e,4
*/
    SingleOutputStreamOperator<CartInfo> cartInfo = inputStream
            //1.过滤数据
            .filter(line -> null != line && line.trim().length() > 0)
            //2. 解析数据,存储至JavaBean对象
            .map(new MapFunction<String, CartInfo>() {
                @Override
                public CartInfo map(String line) throws Exception {
                    String[] arr = line.trim().split(",");
                    return new CartInfo(arr[0], Integer.parseInt(arr[1]));
                }
            });
    //TODO :按照 信号灯编号分组 再进行窗口设置 最后对窗口中的数据进行聚合统计
    SingleOutputStreamOperator<CartInfo> windowDataStream = cartInfo
            .keyBy("sensorId")
            //窗口设置
            .timeWindow(Time.seconds(5))
            //对窗口中数据进行聚合统计
            .sum("count");
            windowDataStream.printToErr();
            env.execute(StreamTumblingTimeWindow_1.class.getSimpleName());
}
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class CartInfo {
        private String sensorId;
        private Integer count;
    }
}